package com.atguigu.ordertimeout_detect

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
 * 双流join
 * 合并两条流(使用join)
 *
 * Project: UserBehaviorAnalysis
 * Package: com.atguigu.ordertimeout_detect
 * Version: 1.0
 *
 * Created by  WangJX  on 2019/12/15 15:18
 */
object TxMatchWithJoin {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val path1: String = TxMatchWithJoin.getClass.getClassLoader.getResource("OrderLog.csv").getPath

    //第一条流
    val orderStream = env.readTextFile(path1)
      .map(
        data => {
          val dataArrays: Array[String] = data.split(",")
          OrderEvent(dataArrays(0).toLong, dataArrays(1), dataArrays(2), dataArrays(3).toLong)
        }
      )
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[OrderEvent](Time.seconds(1)) {
        override def extractTimestamp(element: OrderEvent): Long = element.eventTime * 1000L
      })
      .filter(_.eventType == "pay")
      .keyBy(_.pyId)


    val path2: String = TxMatch.getClass.getClassLoader.getResource("ReceiptLog.csv").getPath
    //第二条流
    val receiptLogStream = env.readTextFile(path2)
      .map(
        data => {
          val dataArrays: Array[String] = data.split(",")
          ReceiptEvent(dataArrays(0), dataArrays(1), dataArrays(2).toLong)
        }
      )
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[ReceiptEvent](Time.seconds(1)) {
        override def extractTimestamp(element: ReceiptEvent): Long = element.eventTime * 1000L
      })
      .keyBy(_.tyId)

    //使用intervalJoin进行合并2条流
    val ouput: DataStream[(OrderEvent, ReceiptEvent)] = orderStream.intervalJoin(receiptLogStream)
      .between(Time.seconds(-5), Time.seconds(3))
      .process(new TxPayMatchByJoin())


    ouput.print("output")

    env.execute("join job")
  }
}

case class ReceiptEvent(
                       tyId: String,
                       eventType: String,
                       eventTime: Long
                       )

class TxPayMatchByJoin() extends ProcessJoinFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]{

  override def processElement(left: OrderEvent, right: ReceiptEvent, ctx: ProcessJoinFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
    out.collect(left, right)
  }
}